package com.mango.ch08;

import java.io.IOException;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import com.mango.HDFSTools.HDFSAPI;
import scala.Tuple2;

public class Spark_ComFriendsJob {

	static Path inputPath = new Path("/Spark_ComFriendsJob/input");// input
	static Path outPath = new Path("/Spark_ComFriendsJob/output");

	@SuppressWarnings({ "serial", "unused", "resource" })
	public static void main(String[] args) {
		// TODO Auto-generated method stub
		// 创建输入文件夹
		Configuration conf = new Configuration();
		try {
			HDFSAPI hdfs = new HDFSAPI(conf);
			// hdfs.createDirectory(transactions);
			// 先删除已经有的输出文件夹
			hdfs.delDirectory(outPath);
			hdfs.closeFilseSystem();
			// hdfs.orpOver();
		} catch (IOException e1) {
			// TODO Auto-generated catch block
			e1.printStackTrace();
			System.out.println("----------文件操作失败");
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (URISyntaxException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		SparkConf sconf = new SparkConf();
		sconf.setAppName("Spark_ComFriendsJob");
		// 建立一个快速串化器
		sconf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer");
		// 现在默认为32mb的缓冲区而不是64mb
		sconf.set("spark.Kryoserializer.buffer.mb", "32");
		JavaSparkContext jsc = new JavaSparkContext(sconf);
		JavaRDD<String> sigleVal = jsc.textFile(inputPath.toString(), 1);
		// 创建需求格式的键值对
		JavaPairRDD<Tuple2<Long, Long>, List<Long>> pairs = sigleVal
				.flatMapToPair(new PairFlatMapFunction<String, Tuple2<Long, Long>, List<Long>>() {

					@Override
					public Iterator<Tuple2<Tuple2<Long, Long>, List<Long>>> call(String t) throws Exception {
						// TODO Auto-generated method stub
						String[] tokens = t.split(",");
						Long person = Long.valueOf(tokens[0]);
						String friendString = tokens[1];// 默认第一个值为person的id
						String[] friendsTokenized = friendString.split(" ");
						if (friendString.length() == 1) {
							// 若length为1表示没有共同好友
							Tuple2<Long, Long> key = buildSortedTuple(person, Long.valueOf(friendsTokenized[0]));
							return Arrays.asList(new Tuple2<Tuple2<Long, Long>, List<Long>>(key, new ArrayList<>()))
									.iterator();
						}
						// 否则将全部朋友遍历添加进来
						List<Long> friends = new ArrayList<Long>();
						for (String s : friendsTokenized) {
							friends.add(Long.valueOf(s));
						}
						// 生成结果集 格式为 ((u1,u2),(a,b,c,d,...))
						List<Tuple2<Tuple2<Long, Long>, List<Long>>> result = new ArrayList<>();
						for (Long friend : friends) {
							Tuple2<Long, Long> key = buildSortedTuple(person, friend);
							result.add(new Tuple2<Tuple2<Long, Long>, List<Long>>(key, friends));
						}
						return result.iterator();
					}
				});
		// 规约数据
		JavaPairRDD<Tuple2<Long, Long>, Iterable<List<Long>>> value = pairs.groupByKey();
		// 查找共同好友
		JavaPairRDD<Tuple2<Long, Long>, Iterable<List<Long>>> results = value
				.mapValues(new Function<Iterable<List<Long>>, // 第一个参数是输入的参数类型
						Iterable<List<Long>>>// 第二个参数是输入的值的类型,不需要再指定建的类型
		() {

					@SuppressWarnings("unchecked")
					@Override
					public Iterable<List<Long>> call(Iterable<List<Long>> v1) throws Exception {
						// TODO Auto-generated method stub
						Map<Long, Integer> counCommon = new HashMap<>();
						int size = 0;
						// 遍历同一key对应的结果集列表
						for (Iterable<Long> iter : v1) {
							size++;// 用来统计当前key有几个结果集列表 ru [[]] [[a,b],[b]] 有两个
							List<Long> list = iterableToList(iter);// 将当前遍历器中的元素转换为list（不一定有元素）
							if (list == null || list.isEmpty()) {// 如果当前遍历次序的list的没有数据，则进入下一次循环
								continue;
							}
							for (Long f : list) {// 如果有数据，list非空，则存入map中，方便在后面求交集，若朋友列表list数据为[100,300,500,600,]
								// 则格式为[（100，1），（300，1），（500，1），（600，1）]
								Integer count = counCommon.get(f);// 朋友不再map中，则加入，若已存在，则将计数器加1
								if (count == null)
									counCommon.put(f, 1);
								else
									counCommon.put(f, ++count);
							}
						}
						// 如果countCommon.Entry<f.count>==countCommon.Entry<f.si.size()>
						// 则有一个共同好友=>因为size记录了 当前key一共有多少个结果集，表示有几个结果集来求交集数据=》即有几个人的好友列表来求共同好友
						// 而map中的值是遍历了每一个结果集，所以当结果集中某个键的count==size时，表示map对应的这个key这个好友是大家共有的
						List<Long> finalCommonFriends = new ArrayList<Long>();// 记录共同好友的列表
						for (Entry<Long, Integer> entry : counCommon.entrySet()) {
							System.out.println("此时的map中的数据为" + entry.getKey() + ":" + entry.getValue());
							if (entry.getValue() == size)
								finalCommonFriends.add(entry.getKey());// 是共同好友加入最终结果列表中
						}
						return (Iterable<List<Long>>) finalCommonFriends.iterator();
					}
				});
		// 查找共同好友 方案二
		JavaPairRDD<Tuple2<Long, Long>, List<Long>> results2 = pairs
				.reduceByKey(new Function2<List<Long>, List<Long>, List<Long>>() {

					@Override
					public List<Long> call(List<Long> v1, List<Long> v2) throws Exception {
						// TODO Auto-generated method stub
						Set<Long> x = new HashSet<>(v1);
						Set<Long> intersection = new HashSet<>();
						for (long entry : v2) {
							if (x.contains(entry))
								intersection.add(entry);
						}
						List<Long> list = new ArrayList<>();
						for (Long l : intersection) {
							list.add(l);
						}
						return list;
					}
				});
		results2.saveAsTextFile(outPath.toString() + "/result");
		Map<Tuple2<Long, Long>, List<Long>> finalResult = results2.collectAsMap();
		for (Entry<Tuple2<Long, Long>, List<Long>> entry : finalResult.entrySet()) {
			System.out.println(entry.getKey() + ":" + entry.getValue());
		}
	}

	protected static List<Long> iterableToList(Iterable<Long> iter) {
		// TODO Auto-generated method stub
		List<Long> list = new ArrayList<>();
		Iterator<Long> it = iter.iterator();
		while (it.hasNext()) {
			list.add(it.next());
		}
		return list;
	}

	protected static Tuple2<Long, Long> buildSortedTuple(Long p1, Long p2) {
		// TODO Auto-generated method stub
		if (p1 > p2) {
			return new Tuple2<Long, Long>(p2, p1);
		} else
			return new Tuple2<Long, Long>(p1, p2);
	}

}
